package pl.com.qapps.datastore.berkeley.impl;

import java.util.Arrays;
import java.util.Comparator;

import pl.com.qapps.datastore.berkeley.BerkeleyUtils;
import pl.com.qapps.datastore.berkeley.ByteKeyComparator;

import com.sleepycat.je.Cursor;
import com.sleepycat.je.CursorConfig;
import com.sleepycat.je.Database;
import com.sleepycat.je.DatabaseEntry;
import com.sleepycat.je.LockMode;
import com.sleepycat.je.OperationStatus;
import com.sleepycat.je.Transaction;

abstract class AncestorRangeScan {
	private final static LockMode mode = LockMode.DEFAULT;
	private final static DatabaseEntry KEY_ONLY = new DatabaseEntry();
	private final static Comparator<byte[]> COMPARATOR = new ByteKeyComparator();

	static {
		KEY_ONLY.setPartial(0, 0, true);
	}

	private final DatabaseEntry firstKey;
	private final DatabaseEntry lastKey;
	private final boolean firstInclusive;
	private final boolean lastInclusive;
	// TODO implement sort duplicates scans
	private final boolean sortDuplicates;
	private final long skip;

	AncestorRangeScan(DatabaseEntry firstKey, DatabaseEntry lastKey,
			boolean firstInclusive, boolean lastInclusive) {
		this(firstKey, lastKey, firstInclusive, lastInclusive, false, 1L);
	}

	AncestorRangeScan(DatabaseEntry firstKey, DatabaseEntry lastKey,
			boolean firstInclusive, boolean lastInclusive, long skip) {
		this(firstKey, lastKey, firstInclusive, lastInclusive, false, skip);
	}

	AncestorRangeScan(DatabaseEntry firstKey, DatabaseEntry lastKey,
			boolean firstInclusive, boolean lastInclusive,
			boolean sortDuplicates) {
		this(firstKey, lastKey, firstInclusive, lastInclusive, sortDuplicates,
				1L);
	}

	AncestorRangeScan(DatabaseEntry firstKey, DatabaseEntry lastKey,
			boolean firstInclusive, boolean lastInclusive,
			boolean sortDuplicates, long skip) {
		this.firstKey = firstKey;
		this.lastKey = lastKey;
		this.firstInclusive = firstInclusive;
		this.lastInclusive = lastInclusive;
		this.sortDuplicates = sortDuplicates;
		this.skip = skip;
	}

	/**
	 * If performed on secondary database, key is the secondary key and data is
	 * data from primary database. For now there is no way to obtain primary
	 * while processing, unless you obtain it from the primary data or secondary
	 * database.
	 * 
	 * @param key
	 *            the primary or secondary key
	 * @param data
	 *            always the data from primary database or null if setKeysOnly(
	 *            {@code true})
	 * @return if continue to find matches, false otherwise
	 */
	protected abstract boolean nextMatch(final DatabaseEntry key,
			final DatabaseEntry data);

	/**
	 * public final void setKeysOnly(boolean keysOnly) { data.setPartial(0, 0,
	 * Boolean.TRUE); }
	 * 
	 * public final boolean isKeyOnly() { return data.getPartial(); }
	 * 
	 * /**
	 * 
	 * @param db
	 * @param firstKey
	 *            can be {@code null}
	 * @param lastKey
	 *            can be null
	 * @param firstInclusive
	 * @param lastInclusive
	 * @param sortDuplicates
	 * @return the matched keys number
	 */
	final long scan(Database db, Transaction txn) {

		// TODO needs optimization, cause of ineffective conditional statements,
		// compare is called twice and almost always twice
		if (firstKey != null && lastKey != null) {
			if (compare(firstKey, lastKey) > 0)
				throw new IllegalArgumentException(
						"firstKey cannot be greater than lastKey");
			if (compare(firstKey, lastKey) == 0
					&& (!firstInclusive || !lastInclusive))
				throw new IllegalArgumentException(
						"there will be no results, keys are equal and at least one of them is excluded from results");
		} else if (firstKey == null && lastKey == null)
			throw new IllegalArgumentException(
					"firstKey and lastKey cannot be both null");

		DatabaseEntry data = new DatabaseEntry();
		DatabaseEntry key = new DatabaseEntry();
		long skipped = 0;

		Cursor cursor = null;
		OperationStatus result = null;

		System.out.println("scanning...  " + this);
		// find first result
		try {
			cursor = db.openCursor(txn,
					new CursorConfig().setReadCommitted(true));
			if (firstKey == null) {
				// TODO don't know if it works
				System.out.println("	no first key, starting from beggining");
				result = cursor.getFirst(key, data, mode);
			} else {
				key.setData(firstKey.getData());
				result = cursor.getSearchKeyRange(key, data, mode);
				if (!firstInclusive && result == OperationStatus.SUCCESS) {
					while (key.equals(firstKey)) {
						// TODO why there is UNCOMMITED lock mode? or was?
						// why omit the result status since we don't care
						cursor.getNext(key, KEY_ONLY, LockMode.READ_UNCOMMITTED);
						System.out
								.println("        omitting excluded first key: "
										+ Arrays.toString(key.getData()));
					}
					// added result =
					result = cursor.getCurrent(key, data, mode);
				}
			}
			if (result == OperationStatus.SUCCESS) {
				System.out.println("	starting from key: "
						+ Arrays.toString(key.getData()));
			} else {
				System.out.println("	key not found, aborting...");

				BerkeleyUtils.dbDump(db, true);
			}
			if (result == OperationStatus.SUCCESS && skip > 0) {
				// TODO why there is UNCOMMITED lock mode? or was?
				skipped = cursor.skipNext(skip, key, data,
						LockMode.READ_UNCOMMITTED);
			}
			// long count = 0;
			if (lastKey == null) {
				// TODO that loop needs some changes, its ugly
				do {
					if (result == OperationStatus.SUCCESS) {
						// count++;
						if (!matchFound(key, data))
							break;
						cursor.getNext(key, data, mode);
					} else {
						break;
					}
				} while (result == OperationStatus.SUCCESS);
			} else {
				if (lastInclusive) {
					while (compare(key, lastKey) < 1) {
						if (result == OperationStatus.SUCCESS) {
							// count++;
							if (!matchFound(key, data))
								break;
							result = cursor.getNext(key, data, mode);
						} else {
							break;
						}
					}
				} else {
					while (compare(key, lastKey) < 0) {
						if (result == OperationStatus.SUCCESS) {
							// count++;
							if (!matchFound(key, data))
								break;
							result = cursor.getNext(key, data, mode);
						} else {
							break;
						}
					}
				}
			}
			return skipped;
			// } catch(Exception e) {
			// e.printStackTrace();
			// return 0;
		} finally {
			System.out.println("scanning done");
			// for clarity
			BerkeleyUtils.dbDump(db);
			if (cursor != null)
				cursor.close();
		}
	}

	private final boolean matchFound(DatabaseEntry key, DatabaseEntry data) {
		System.out.println("	match found, key="
				+ Arrays.toString(key.getData()));
		return nextMatch(key, data);
	}

	@Override
	public String toString() {
		return "RangeScan"
				+ (firstInclusive ? "<" : "(")
				+ (firstKey != null ? Arrays.toString(firstKey.getData()) : "&")
				+ ";"
				+ (lastKey != null ? Arrays.toString(lastKey.getData()) : "&")
				+ (lastInclusive ? ">" : ")");
		// return "RangeScan"
		// + (firstInclusive ? "<" : "(")
		// + (firstKey != null ? new String(firstKey.getData(), 0,
		// firstKey.getSize() - 1) : "&")
		// + ";"
		// + (lastKey != null ? new String(lastKey.getData(), 0,
		// lastKey.getSize() - 1)
		// + "" : "&") + (lastInclusive ? ">" : ")");
	}

	private static int compare(byte[] first, byte[] second) {
		// return ByteBuffer.wrap(first).compareTo(ByteBuffer.wrap(second));
		return COMPARATOR.compare(first, second);
	}

	private static int compare(DatabaseEntry first, DatabaseEntry second) {
		return compare(first.getData(), second.getData());
	}
}
